package com.sailors.common.task.sdk.core.handle;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;

public abstract class KafkaProducer<T> {

    protected static Properties props;

    protected static Producer<String, String> producer;
    /**
     * OFF
     */
    private static final AtomicBoolean OFF = new AtomicBoolean(false);

    public static void initConfig(JSONObject config) {
        // 只能启动一次
        if (OFF.compareAndSet(false, true)) {
            Properties props = new Properties();
            props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, config.get(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG));
            // 至少要等待leader成功将数据写入
            props.put(ProducerConfig.ACKS_CONFIG, config.getOrDefault(ProducerConfig.ACKS_CONFIG, "1"));
            //#重试三次
            props.put(ProducerConfig.RETRIES_CONFIG, config.getOrDefault(ProducerConfig.RETRIES_CONFIG, 3));
            // 每隔300毫秒重试一次
            props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, config.getOrDefault(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300));
            //设置本地缓存区大小32MB
            props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, config.getOrDefault(ProducerConfig.BUFFER_MEMORY_CONFIG, 1024 * 1024 * 32));
            // 缓存超过16kb就发一批
            props.put(ProducerConfig.BATCH_SIZE_CONFIG, config.getOrDefault(ProducerConfig.BATCH_SIZE_CONFIG, 16 * 1024));
            // 等待时间超过10毫秒不管有没有满16k都发一次
            props.put(ProducerConfig.LINGER_MS_CONFIG, config.getOrDefault(ProducerConfig.LINGER_MS_CONFIG, 10));
            props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
            KafkaProducer.props = props;
            KafkaProducer.producer = new org.apache.kafka.clients.producer.KafkaProducer(props);
        }
    }
}
